package com.service;

import com.config.RabbitMQConfig;
import com.dao.ChatMessageDao;
import com.entity.ChatMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;

import com.rabbitmq.client.Channel;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * 消息消费者
 */
@Service
public class MessageConsumer {

    private static final Logger logger = LoggerFactory.getLogger(MessageConsumer.class);

    @Autowired
    private ChatMessageDao chatMessageDao;

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 处理聊天消息
     * @param message 消息对象
     * @param channel RabbitMQ通道
     * @param deliveryTag 消息标签
     */
    @RabbitListener(queues = "#{T(com.config.RabbitMQConfig).USER_QUEUE_PREFIX + 'default'}")
    public void handleChatMessage(ChatMessage message, Channel channel, 
                                 @org.springframework.amqp.support.AmqpHeaders.Message Message amqpMessage) {
        long deliveryTag = amqpMessage.getMessageProperties().getDeliveryTag();
        String receiverOpenId = message.getReceiverOpenid();
        
        try {
            logger.info("接收到消息: sender={}, receiver={}, content={}", 
                message.getSenderOpenid(), receiverOpenId, message.getContent());

            // 1. 检查消息是否已处理（幂等性）
            String messageKey = "processed_message:" + message.getId();
            if (redisTemplate.hasKey(messageKey)) {
                logger.warn("消息已处理，跳过: messageId={}", message.getId());
                channel.basicAck(deliveryTag, false);
                return;
            }

            // 2. 存储消息到数据库
            try {
                chatMessageDao.insert(message);
                logger.info("消息存储成功: messageId={}", message.getId());
            } catch (Exception e) {
                logger.error("消息存储失败: messageId={}, error={}", message.getId(), e.getMessage(), e);
                // 存储失败，拒绝消息并重新入队
                channel.basicNack(deliveryTag, false, true);
                return;
            }

            // 3. 检查用户是否在线
            boolean isOnline = isUserOnline(receiverOpenId);
            if (isOnline) {
                // 4. 通过WebSocket推送消息
                try {
                    messagingTemplate.convertAndSendToUser(
                        receiverOpenId,
                        "/queue/messages",
                        message
                    );
                    logger.info("WebSocket推送成功: receiver={}", receiverOpenId);
                } catch (Exception e) {
                    logger.error("WebSocket推送失败: receiver={}, error={}", receiverOpenId, e.getMessage(), e);
                    // WebSocket推送失败，但消息已存储，仍然确认消息
                }
            } else {
                logger.info("用户离线，消息已存储: receiver={}", receiverOpenId);
            }

            // 5. 标记消息为已处理（幂等性）
            redisTemplate.opsForValue().set(messageKey, "processed", 24, TimeUnit.HOURS);

            // 6. 手动确认消息
            channel.basicAck(deliveryTag, false);
            logger.info("消息处理完成: messageId={}", message.getId());

        } catch (Exception e) {
            logger.error("消息处理异常: messageId={}, error={}", message.getId(), e.getMessage(), e);
            try {
                // 处理失败时拒绝消息并重新入队
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ioException) {
                logger.error("拒绝消息失败: {}", ioException.getMessage(), ioException);
            }
        }
    }

    /**
     * 处理死信消息
     * @param message 消息对象
     * @param channel RabbitMQ通道
     * @param deliveryTag 消息标签
     */
    @RabbitListener(queues = "#{T(com.config.RabbitMQConfig).DLX_QUEUE_PREFIX + 'default'}")
    public void handleDeadLetterMessage(ChatMessage message, Channel channel, 
                                       @org.springframework.amqp.support.AmqpHeaders.Message Message amqpMessage) {
        long deliveryTag = amqpMessage.getMessageProperties().getDeliveryTag();
        
        try {
            logger.warn("处理死信消息: messageId={}, sender={}, receiver={}", 
                message.getId(), message.getSenderOpenid(), message.getReceiverOpenid());

            // 死信消息处理逻辑
            // 可以记录到特殊表或发送通知
            message.setStatus("failed");
            chatMessageDao.insert(message);

            // 确认死信消息
            channel.basicAck(deliveryTag, false);
            logger.info("死信消息处理完成: messageId={}", message.getId());

        } catch (Exception e) {
            logger.error("死信消息处理失败: messageId={}, error={}", message.getId(), e.getMessage(), e);
            try {
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ioException) {
                logger.error("拒绝死信消息失败: {}", ioException.getMessage(), ioException);
            }
        }
    }

    /**
     * 检查用户是否在线
     * @param openId 用户OpenID
     * @return 是否在线
     */
    private boolean isUserOnline(String openId) {
        try {
            String onlineKey = "online_user:" + openId;
            return redisTemplate.hasKey(onlineKey);
        } catch (Exception e) {
            logger.error("检查用户在线状态失败: openId={}, error={}", openId, e.getMessage(), e);
            return false;
        }
    }

    /**
     * 设置用户在线状态
     * @param openId 用户OpenID
     * @param isOnline 是否在线
     */
    public void setUserOnlineStatus(String openId, boolean isOnline) {
        try {
            String onlineKey = "online_user:" + openId;
            if (isOnline) {
                redisTemplate.opsForValue().set(onlineKey, "online", 30, TimeUnit.MINUTES);
            } else {
                redisTemplate.delete(onlineKey);
            }
            logger.info("用户在线状态更新: openId={}, isOnline={}", openId, isOnline);
        } catch (Exception e) {
            logger.error("设置用户在线状态失败: openId={}, isOnline={}, error={}", 
                openId, isOnline, e.getMessage(), e);
        }
    }
}
